package com.example.meeting.config;


import cn.hutool.extra.spring.SpringUtil;
import cn.hutool.json.JSONObject;
import com.example.meeting.model.domain.Meeting;
import com.example.meeting.service.IMeetingService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.*;

@Component
@ServerEndpoint("/websocket/{id}")
@Slf4j
public class MyHandler {
    /**
     * 会议id -> websocket连接列表
     * MyHandler不是单例的，每个websocket连接都会创建一个MyHandler对象
     * 所以这里使用静态变量来保存websocket连接列表
     */
    private static final ConcurrentMap<Long, List<Session>> sessions = new ConcurrentHashMap<>();

    @OnOpen
    public void onOpen(Session session, @PathParam(value = "id") Long id) throws IOException {
        // 检查会议是否存在
        IMeetingService meetingService = SpringUtil.getBean(IMeetingService.class);
        Meeting meeting = meetingService.selectMeetingById(id);
        if (meeting == null) {
            log.info("websocket连接失败 id为{}的会议不存在", id);
            JSONObject msg = new JSONObject();
            msg.set("type", "noSuchMeeting");
            session.getBasicRemote().sendText(msg.toString());
            session.close();
            return;
        }

        // 加入websocket连接表
        log.info("websocket连接成功 会议id:{}", id);
        sessions.merge(id, new CopyOnWriteArrayList<Session>() {{
            add(session);
        }}, (oldValue, newValue) -> {
            oldValue.add(session);
            return oldValue;
        });
//        log.info("当前websocket连接表:{}", sessions);

        // 发送会议信息
        JSONObject msg = new JSONObject();
        msg.set("type", "meetingInfo");
        msg.set("data", meeting);
        session.getBasicRemote().sendText(msg.toString());
    }

    @OnClose
    public void onClose(Session session, @PathParam(value = "id") Long id) {
        log.info("websocket连接关闭 会议id:{}", id);
        sessions.getOrDefault(id, Collections.emptyList()).remove(session);
//        log.info("当前websocket连接表:{}", sessions);
    }

    @OnMessage
    public void onMessage(String message, @PathParam(value = "id") Long id) {
        log.info("id为{}的websocket客户端发送消息: {}", id, message);
    }

    @OnError
    public void onError(Session session, Throwable error) {
        log.error("websocket发生错误: {}", error.getMessage());
    }

    // a single thread pool to handle the message sending, with an unbounded queue
    private static ThreadPoolExecutor messageSender = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>());

    public static void sendMessage(Long id, String message) {
        for (Session session : sessions.getOrDefault(id, Collections.emptyList())) {
            messageSender.execute(() -> {
                try {
                    session.getBasicRemote().sendText(message);
                    log.info("发送消息成功, 接收id: {}, 消息: {}", id, message);
                } catch (IOException e) {
                    log.error("发送消息失败, 原因: {}", e.getMessage());
                }
            });
        }
    }

    public static void broadcastMessage(String message) {
        for (Long id : sessions.keySet()) {
            for (Session session : sessions.get(id)) {
                messageSender.execute(() -> {
                    try {
                        session.getBasicRemote().sendText(message);
                    } catch (IOException e) {
                        log.error("发送消息失败, 原因: {}", e.getMessage());
                    }
                });
            }
            log.info("提交广播消息成功, 接收id: {}, 消息: {}", id, message);
        }
    }

}
